﻿#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <stdarg.h>
#include <unistd.h>
#include <sys/time.h>
#include <time.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <arpa/inet.h>
#include "ngx_c_conf.h"
#include "ngx_macro.h"
#include "ngx_global.h"
#include "ngx_func.h"
#include "ngx_c_socket.h"
#include "ngx_c_memory.h"
#include "ngx_c_lockmutex.h"
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      构造函数
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
CSocekt::CSocekt()
{
    m_worker_connections = 1;           //epoll连接最大项数
    m_ListenPortCount = 1;              //监听一个端口
    m_RecyConnectionWaitTime = 60;      //等待这么些秒后才回收连接

    m_epollhandle = -1;                 //epoll返回的句柄

    m_iLenPkgHeader = sizeof(COMM_PKG_HEADER);    //包头的sizeof值
    m_iLenMsgHeader =  sizeof(STRUC_MSG_HEADER);  //消息头的sizeof值

    //各种队列相关
    m_iSendMsgQueueCount     = 0;     //发消息队列大小
    m_totol_recyconnection_n = 0;     //待释放连接队列大小
    m_cur_size_              = 0;     //当前计时队列尺寸
    m_timer_value_           = 0;     //当前计时队列头部的时间值
    m_iDiscardSendPkgCount   = 0;     //丢弃的发送数据包数量

    //在线用户相关
    m_onlineUserCount        = 0;     //在线用户数量统计，先给0
    m_lastprintTime          = 0;     //上次打印统计信息的时间，先给0
    return; 
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      套接字初始化(fork()子进程之前干这个事)
 * @author     txj
 * @version    v1
 * @return     成功返回true，失败返回false
 * @date       2021/7/29
 **************************************************************/
bool CSocekt::Initialize()
{
    ReadConf();//读配置项
    if(ngx_open_listening_sockets() == false)  //打开监听端口
    {
        return false;
    } 
    return true;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      子进程中才需要执行的初始化函数
 * @author     txj
 * @version    v1
 * @return     成功返回true，失败返回false
 * @date       2021/7/29
 **************************************************************/
bool CSocekt::Initialize_subproc()
{
    /*发消息互斥量初始化*/
    if(pthread_mutex_init(&m_sendMessageQueueMutex, NULL)  != 0)
    {        
        ngx_log_stderr(0,"CSocekt::Initialize_subproc()中\
            pthread_mutex_init(&m_sendMessageQueueMutex)失败.");
        return false;
    }
    /*连接相关互斥量初始化*/
    if(pthread_mutex_init(&m_connectionMutex, NULL)  != 0)
    {
        ngx_log_stderr(0,"CSocekt::Initialize_subproc()中\
            pthread_mutex_init(&m_connectionMutex)失败.");
        return false;
    }
    //连接回收队列相关互斥量初始化
    if(pthread_mutex_init(&m_recyconnqueueMutex, NULL)  != 0)
    {
        ngx_log_stderr(0,"CSocekt::Initialize_subproc()中\
            pthread_mutex_init(&m_recyconnqueueMutex)失败.");
        return false;
    }
    //和时间处理队列有关的互斥量初始化
    if(pthread_mutex_init(&m_timequeueMutex, NULL)  != 0)
    {
        ngx_log_stderr(0,"CSocekt::Initialize_subproc()中\
            pthread_mutex_init(&m_timequeueMutex)失败.");
        return false;
    }
    /*信号量*/
    if(sem_init(&m_semEventSendQueue,0,0) == -1)
    {
        ngx_log_stderr(0,"CSocekt::Initialize_subproc()中\
            sem_init(&m_semEventSendQueue,0,0)失败.");
        return false;
    }

    //创建线程，专门用来发送数据的线程
    int err;
    ThreadItem *pSendQueue;
    m_threadVector.push_back(pSendQueue = new ThreadItem(this)); 
    /*创建线程，错误不返回到errno，一般返回错误码*/
    err = pthread_create(&pSendQueue->_Handle, NULL, 
            ServerSendQueueThread,pSendQueue); 
    if(err != 0)
    {
        ngx_log_stderr(0,"CSocekt::Initialize_subproc()中\
            pthread_create(ServerSendQueueThread)失败.");
        return false;
    }

    ThreadItem *pRecyconn;    //专门用来回收连接的线程
    m_threadVector.push_back(pRecyconn = new ThreadItem(this)); 
    err = pthread_create(&pRecyconn->_Handle, NULL, 
            ServerRecyConnectionThread,pRecyconn);
    if(err != 0)
    {
        ngx_log_stderr(0,"CSocekt::Initialize_subproc()中\
            pthread_create(ServerRecyConnectionThread)失败.");
        return false;
    }

    if(m_ifkickTimeCount == 1)  //是否开启踢人时钟，1：开启   0：不开启
    {
        ThreadItem *pTimemonitor;    //专门用来处理到期不发心跳包的用户踢出的线程
        m_threadVector.push_back(pTimemonitor = new ThreadItem(this)); 
        err = pthread_create(&pTimemonitor->_Handle, NULL, 
                ServerTimerQueueMonitorThread,pTimemonitor);
        if(err != 0)
        {
            ngx_log_stderr(0,"CSocekt::Initialize_subproc()中\
                pthread_create(ServerTimerQueueMonitorThread)失败.");
            return false;
        }
    }
    return true;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      析构函数
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
CSocekt::~CSocekt()
{
    //监听端口相关内存的释放
    std::vector<lpngx_listening_t>::iterator pos;
    for(pos = m_ListenSocketList.begin(); pos != m_ListenSocketList.end(); ++pos)
    {       
        delete (*pos);
    }
    m_ListenSocketList.clear();
    return;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      关闭退出函数[子进程中执行]
 * @author     txj
 * @version    v1
 * @return     void
 * @date       2021/7/29
 **************************************************************/
void CSocekt::Shutdown_subproc()
{
    if(sem_post(&m_semEventSendQueue)==-1)
    {
        ngx_log_stderr(0,"CSocekt::Shutdown_subproc()中\
            sem_post(&m_semEventSendQueue)失败.");
    }

    std::vector<ThreadItem*>::iterator iter;
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        pthread_join((*iter)->_Handle, NULL);
    }

    //释放一下new出来的ThreadItem线程池中的线程
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        if(*iter)
            delete *iter;
    }
    m_threadVector.clear();

    //队列相关
    clearMsgSendQueue();
    clearconnection();
    clearAllFromTimerQueue();

    //多线程相关
    pthread_mutex_destroy(&m_connectionMutex);          //连接相关互斥量释放
    pthread_mutex_destroy(&m_sendMessageQueueMutex);    //发消息互斥量释放
    pthread_mutex_destroy(&m_recyconnqueueMutex);       //连接回收队列相关的互斥量释放
    pthread_mutex_destroy(&m_timequeueMutex);           //时间处理队列相关的互斥量释放
    sem_destroy(&m_semEventSendQueue);                  //发消息相关线程信号量释放
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      清理TCP发送消息队列
 * @author     txj
 * @version    v1
 * @return     void
 * @date       2021/7/29
 **************************************************************/
void CSocekt::clearMsgSendQueue()
{
    char *sTmpMempoint;
    CMemory *p_memory = CMemory::GetInstance();
    while(!m_MsgSendQueue.empty())
    {
        sTmpMempoint = m_MsgSendQueue.front();
        m_MsgSendQueue.pop_front(); 
        p_memory->FreeMemory(sTmpMempoint);
    }
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      专门用于读各种配置项
 * @author     txj
 * @version    v1
 * @return     void
 * @date       2021/7/29
 **************************************************************/
void CSocekt::ReadConf()
{
    /*单例类*/
    CConfig *p_config = CConfig::GetInstance();
    /*epoll连接的最大项数*/
    m_worker_connections      = p_config->GetIntDefault("worker_connections",m_worker_connections);
    /*要监听的端口数量*/
    m_ListenPortCount         = p_config->GetIntDefault("ListenPortCount",m_ListenPortCount);
    /*等待这么些秒后才回收连接*/
    m_RecyConnectionWaitTime  = p_config->GetIntDefault("Sock_RecyConnectionWaitTime",m_RecyConnectionWaitTime); 

    /*是否开启踢人时钟，1：开启   0：不开启*/
    m_ifkickTimeCount         = p_config->GetIntDefault("Sock_WaitTimeEnable",0);

    /*多少秒检测一次是否心跳超时，只有当Sock_WaitTimeEnable = 1时，本项才有用*/
    m_iWaitTime               = p_config->GetIntDefault("Sock_MaxWaitTime",m_iWaitTime);

    //不建议低于5秒钟，因为无需太频繁
    m_iWaitTime               = (m_iWaitTime > 5)?m_iWaitTime:5; 

    /*当时间到达Sock_MaxWaitTime指定的时间时，直接把客户端踢出去，
    只有当Sock_WaitTimeEnable = 1时，本项才有用*/
    m_ifTimeOutKick           = p_config->GetIntDefault("Sock_TimeOutKick",0); 

    /*Flood攻击检测是否开启,1：开启   0：不开启*/
    m_floodAkEnable          = p_config->GetIntDefault("Sock_FloodAttackKickEnable",0); 

    /*表示每次收到数据包的时间间隔是100(毫秒)*/
    m_floodTimeInterval      = p_config->GetIntDefault("Sock_FloodTimeInterval",100);
    /*累积多少次踢出此人*/
    m_floodKickCount         = p_config->GetIntDefault("Sock_FloodKickCounter",10); 
    return;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      监听端口(支持多个端口,在创建worker进程之前就要执行这个函数)
 * @author     txj
 * @version    v1
 * @return     false/true
 * @date       2021/7/29
 **************************************************************/
bool CSocekt::ngx_open_listening_sockets()
{
    int                isock;                //socket
    struct sockaddr_in serv_addr;            //服务器的地址结构体
    int                iport;                //端口
    char               strinfo[100];         //临时字符串 

    memset(&serv_addr,0,sizeof(serv_addr)); 
    serv_addr.sin_family = AF_INET;
    /*监听本地所有的IP地址；INADDR_ANY表示的是一个服务器上
    所有的网卡（服务器可能不止一个网卡）多个本地ip地址都进行绑定端口号，进行侦听*/
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    /*因为要读取配置项，所以这里要读取这个单例类*/
    CConfig *p_config = CConfig::GetInstance();
    for(int i = 0; i < m_ListenPortCount; i++)
    {
        isock = socket(AF_INET,SOCK_STREAM,0);
        if(isock == -1)
        {
            ngx_log_stderr(errno,"CSocekt::Initialize()中socket()失败,i=%d.",i);
            return false;
        }

        /*设置套接字选项，解决time_wait问题*/
        int reuseaddr = 1;
        if(setsockopt(isock,SOL_SOCKET, SO_REUSEADDR,
                (const void *)&reuseaddr, sizeof(reuseaddr)) == -1)
        {
            ngx_log_stderr(errno,"CSocekt::Initialize()中setsockopt(SO_REUSEADDR)\
                            失败,i=%d.",i);
            close(isock);
            return false;
        }
        /*为处理惊群问题使用reuseport，端口复用需要内核支持*/
        int reuseport = 1;
        if(setsockopt(isock, SOL_SOCKET, SO_REUSEPORT,
                (const void *) &reuseport, sizeof(int)) == -1)
        {
            ngx_log_stderr(errno,"CSocekt::Initialize()中\
                setsockopt(SO_REUSEPORT)失败",i);
        }
        /*设置该socket为非阻塞*/
        if(setnonblocking(isock) == false)
        {
            ngx_log_stderr(errno,"CSocekt::Initialize()中\
                setnonblocking()失败,i=%d.",i);
            close(isock);
            return false;
        }
        strinfo[0] = 0;
        /*字符串连接，ListenPort0/ListenPort1*/
        sprintf(strinfo,"ListenPort%d",i);
        /*获取要监听的端口号*/
        iport = p_config->GetIntDefault(strinfo, 8000);
        /*将端口号填入serv_addr结构体*/
        serv_addr.sin_port = htons((in_port_t)iport);
        /*绑定服务器地址结构体*/
        if(bind(isock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) == -1)
        {
            ngx_log_stderr(errno,"CSocekt::Initialize()中bind()失败,i=%d.",i);
            close(isock);
            return false;
        }
        /*开始监听, 半连接和已连接队列之和为NGX_LISTEN_BACKLOG*/
        if(listen(isock,NGX_LISTEN_BACKLOG) == -1)
        {
            ngx_log_stderr(errno,"CSocekt::Initialize()中listen()失败,i=%d.",i);
            close(isock);
            return false;
        }
        /*这里用容器保存下监听套接字的信息*/
        lpngx_listening_t p_listensocketitem = new ngx_listening_t;
        /*清0处理*/
        memset(p_listensocketitem,0,sizeof(ngx_listening_t));
        /*记录下所监听的端口号*/
        p_listensocketitem->port = iport;
        /*监听套接字保存下来*/
        p_listensocketitem->fd   = isock;
        ngx_log_error_core(NGX_LOG_INFO,0,"监听%d端口成功!",iport);
        /*加入到队列中*/
        m_ListenSocketList.push_back(p_listensocketitem);
    }
    /*判断一下监听端口数量，看是否存储错误*/
    if(m_ListenSocketList.size() <= 0)
    {
        return false;
    }
    return true;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      设置socket连接为非阻塞模式
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
bool CSocekt::setnonblocking(int sockfd) 
{
    /*0：清除，1：设置*/
    int nb=1;
    /*FIONBIO：设置/清除非阻塞I/O标记：0：清除，1：设置*/
    if(ioctl(sockfd, FIONBIO, &nb) == -1) 
    {
        return false;
    }
    return true;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      关闭socket，什么时候用，我们现在先不确定，先把这个函数预备在这里
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
void CSocekt::ngx_close_listening_sockets()
{
    for(int i = 0; i < m_ListenPortCount; i++) //要关闭这么多个监听端口
    {  
        ngx_log_stderr(0,"端口是%d,socketid是%d.",
            m_ListenSocketList[i]->port,m_ListenSocketList[i]->fd);
        close(m_ListenSocketList[i]->fd);
        ngx_log_error_core(NGX_LOG_INFO,0,"关闭监听端口%d!",
            m_ListenSocketList[i]->port);
    }
    return;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      将一个待发送消息入到发消息队列中
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
void CSocekt::msgSend(char *psendbuf)
{
    CMemory *p_memory = CMemory::GetInstance();
    CLock lock(&m_sendMessageQueueMutex);

    /*发送消息队列过大也可能给服务器带来风险*/
    if(m_iSendMsgQueueCount > 50000)
    {
        m_iDiscardSendPkgCount++;
        /*直接丢弃这个数据吧*/
        p_memory->FreeMemory(psendbuf);
        return;
    }

    //总体数据并无风险，不会导致服务器崩溃，要看看个体数据，找一下恶意者了
    LPSTRUC_MSG_HEADER pMsgHeader = (LPSTRUC_MSG_HEADER)psendbuf;
    lpngx_connection_t p_Conn = pMsgHeader->pConn;

    /*针对某个客户端累积超过400个包发不出去就关闭*/
    if(p_Conn->iSendCount > 400)
    {
        ngx_log_stderr(0,"CSocekt::msgSend()中发现某用户%d积压了大量待发送数据包，切断与他的连接！",p_Conn->fd);
        /*丢弃包的数量加一*/
        m_iDiscardSendPkgCount++;
        p_memory->FreeMemory(psendbuf);
        zdClosesocketProc(p_Conn);
        return;
    }

    /*发送队列中有的数据条目数+1*/
    ++p_Conn->iSendCount;
    m_MsgSendQueue.push_back(psendbuf);

    /*原子操作*/
    ++m_iSendMsgQueueCount;
    /*信号量加一，发送消息的线程可以往下走了*/
    if(sem_post(&m_semEventSendQueue)==-1)
    {
        ngx_log_stderr(0,"CSocekt::msgSend()中sem_post\
            (&m_semEventSendQueue)失败.");
    }
    return;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      关闭连接
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
void CSocekt::zdClosesocketProc(lpngx_connection_t p_Conn)
{
    if(m_ifkickTimeCount == 1)
    {
        DeleteFromTimerQueue(p_Conn);
    }
    if(p_Conn->fd != -1)
    {
        /*这个socket关闭，关闭后epoll就会被
            从红黑树中删除，所以这之后无法收到任何epoll事件*/
        close(p_Conn->fd);
        p_Conn->fd = -1;
    }
    if(p_Conn->iThrowsendCount > 0)
    {
        --p_Conn->iThrowsendCount;
    }
    inRecyConnectQueue(p_Conn);
    return;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      测试是否flood攻击成立，成立则返回true，否则返回false
 *             如果有恶意用户在连续发包(10次)的时间间隔内小于100毫秒，
 *             就将该用户关闭
 * @author     txj
 * @version    v1
 * @return     false/true
 * @date       2021/7/29
 **************************************************************/
bool CSocekt::TestFlood(lpngx_connection_t pConn)
{
    struct  timeval sCurrTime;
    uint64_t        iCurrTime;
    bool  reco      = false;

    /*取得当前时间*/
    gettimeofday(&sCurrTime, NULL);

    /*将当前时间转化为毫秒*/
    iCurrTime = (sCurrTime.tv_sec * 1000 + sCurrTime.tv_usec / 1000);
    /*如果当前时间减去上一次收到包的时间小于m_floodTimeInterval(100毫秒)*/
    if((iCurrTime - pConn->FloodkickLastTime) < m_floodTimeInterval)
    {
        pConn->FloodAttackCount++;
        pConn->FloodkickLastTime = iCurrTime;
    }
    else
    {
        /*如果不是间隔小于100毫秒内发包则恢复计数值*/
        pConn->FloodAttackCount = 0;
        pConn->FloodkickLastTime = iCurrTime;
    }

    ngx_log_stderr(0,"pConn->FloodAttackCount=%d,m_floodKickCount=%d.",
        pConn->FloodAttackCount,m_floodKickCount);

    /*攻击数大于m_floodKickCount(10次)*/
    if(pConn->FloodAttackCount >= m_floodKickCount)
    {
        /*踢人的标志*/
        reco = true;
    }
    return reco;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      epoll功能初始化子进程中进行本函数被ngx_worker_process_init()所调用
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
int CSocekt::ngx_epoll_init()
{
    m_epollhandle = epoll_create(m_worker_connections);
    if (m_epollhandle == -1)
    {
        ngx_log_stderr(errno,"CSocekt::ngx_epoll_init()中epoll_create()失败.");
        exit(2);
    }
    /*创建连接池*/
    initconnection();
    /*因为之前已经开始监听端口的初始化了，所以这里要从连接池里面扣除连接和监听套接字进行绑定*/
    std::vector<lpngx_listening_t>::iterator pos;
    for(pos = m_ListenSocketList.begin(); pos != m_ListenSocketList.end(); ++pos)
    {
        /*从连接池中获取一个空闲连接对象，(*pos)->fd是监听套接字的句柄*/
        lpngx_connection_t p_Conn = ngx_get_connection((*pos)->fd);
        if (p_Conn == NULL)
        {
            ngx_log_stderr(errno,"CSocekt::ngx_epoll_init()\
                中ngx_get_connection()失败.");
            exit(2);
        }
        /*连接对象 和监听对象关联，方便通过连接对象找监听对象*/
        p_Conn->listening = (*pos);
        /*监听对象 和连接对象关联，方便通过监听对象找连接对象*/
        (*pos)->connection = p_Conn;
        /*对监听端口的读事件设置处理方法*/
        p_Conn->rhandler = &CSocekt::ngx_event_accept;
        if(ngx_epoll_oper_event((*pos)->fd,            //socekt句柄
                                EPOLL_CTL_ADD,         //事件类型，这里是增加
                                EPOLLIN|EPOLLRDHUP,    //标志这里代表要增加的标志,EPOLLIN：可读，EPOLLRDHUP：TCP连接的远端关闭或者半关闭
                                0,                     //对于事件类型为增加的，不需要这个参数
                                p_Conn                 //连接池中的连接
                                ) == -1)
        {
            exit(2);
        }
    }
    return 1;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      
 * @author     txj
 * @version    v1
 * @return     1/-1
 * @date       2021/7/29
 **************************************************************/
int CSocekt::ngx_epoll_oper_event(
                        int                fd,               //句柄，一个socket
                        uint32_t           eventtype,        //事件类型，一般是EPOLL_CTL_ADD，EPOLL_CTL_MOD，EPOLL_CTL_DEL
                        uint32_t           flag,             //标志，具体含义取决于eventtype
                        int                bcaction,         //补充动作，用于补充flag标记的不足  :  0：增加   1：去掉 2：完全覆盖
                        lpngx_connection_t pConn             //pConn：一个连接EPOLL_CTL_ADD时增加到红黑树中去，将来epoll_wait时能取出来用
                        )
{
    struct epoll_event ev;
    memset(&ev, 0, sizeof(ev));
    /*往红黑树中增加节点*/
    if(eventtype == EPOLL_CTL_ADD)
    {
        ev.events = flag;
        pConn->events = flag;
    }
    /*节点已经在红黑树中，修改节点的事件信息*/
    else if(eventtype == EPOLL_CTL_MOD)
    {
        /*先把标记恢复回来*/
        ev.events = pConn->events;
        if(bcaction == 0)
        {
            ev.events |= flag;
        }
        else if(bcaction == 1)
        {
            ev.events &= ~flag;
        }
        else
        {
            ev.events = flag;
        }
        pConn->events = ev.events;
    }
    else
    {
        return 1;
    } 
    /*EPOLL_CTL_MOD的话，这个会覆盖调，这里再赋值一次*/
    ev.data.ptr = (void *)pConn;
    if(epoll_ctl(m_epollhandle, eventtype, fd, &ev) == -1)
    {
        ngx_log_stderr(errno,"CSocekt::ngx_epoll_oper_event()\
            中epoll_ctl(%d,%ud,%ud,%d)失败.",fd,eventtype,flag,bcaction);
        return -1;
    }
    return 1;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      
 * @author     txj
 * @version    v1
 * @return     1/0
 * @date       2021/7/29
 **************************************************************/
int CSocekt::ngx_epoll_process_events(int timer) 
{ 
    int events = epoll_wait(m_epollhandle, m_events, NGX_MAX_EVENTS, timer);
    /*有错误*/
    if(events == -1)
    {
        /*信号中断出去*/
        if(errno == EINTR)
        {
            ngx_log_error_core(NGX_LOG_INFO,errno,
                "CSocekt::ngx_epoll_process_events()中epoll_wait()失败!"); 
            return 1;
        }
        else/*其它情况*/
        {
            ngx_log_error_core(NGX_LOG_ALERT,errno,
                "CSocekt::ngx_epoll_process_events()中epoll_wait()失败!"); 
            return 0;
        }
    }
    /*超时处理*/
    else if(events == 0)
    {
        if(timer != -1)
        {
            return 1;
        }
        return 0;
    }
    else/*正常情况*/
    {
        lpngx_connection_t p_Conn;
        uint32_t           revents;
        for(int i = 0; i < events; ++i)
        {
            /*取出相应的连接池*/
            p_Conn = (lpngx_connection_t)(m_events[i].data.ptr);
            /*取出事件类型*/
            revents = m_events[i].events;

            /*可读事件，对方发送数据过来了*/
            if(revents & EPOLLIN)
            {
                (this->* (p_Conn->rhandler) )(p_Conn);
            }
            /*可写事件，输出缓冲区可写*/
            if(revents & EPOLLOUT)
            {
                if(revents & (EPOLLERR | EPOLLHUP | EPOLLRDHUP))
                {
                    --p_Conn->iThrowsendCount;
                }
            }
            else
            {
                (this->* (p_Conn->whandler) )(p_Conn);
            }
        }
    }
    return 1;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      处理发送消息队列的线程
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
void* CSocekt::ServerSendQueueThread(void* threadData)
{
    ThreadItem *pThread = static_cast<ThreadItem*>(threadData);
    CSocekt *pSocketObj = pThread->_pThis;
    std::list <char *>::iterator pos,pos2,posend;
    LPSTRUC_MSG_HEADER  pMsgHeader;
    LPCOMM_PKG_HEADER   pPkgHeader;
    lpngx_connection_t  p_Conn;
    unsigned short      itmp;
    ssize_t             sendsize;
    char               *pMsgBuf;
    int                 err;

    CMemory *p_memory = CMemory::GetInstance();
    while(g_stopEvent == 0)
    {
        if(sem_wait(&pSocketObj->m_semEventSendQueue) == -1)
        {
            if(errno != EINTR)
                ngx_log_stderr(errno,"CSocekt::ServerSendQueueThread()\
                    中sem_wait(&pSocketObj->m_semEventSendQueue)失败.");
        }
        if(g_stopEvent != 0)
        {
            break;
        }
        /*发送队列里有数据*/
        if(pSocketObj->m_iSendMsgQueueCount > 0) 
        {
            err = pthread_mutex_lock(&pSocketObj->m_sendMessageQueueMutex);
            if(err != 0)
            {
                ngx_log_stderr(err,"CSocekt::ServerSendQueueThread()
                    中pthread_mutex_lock()失败，返回的错误码为%d!",err);
            }

            pos    = pSocketObj->m_MsgSendQueue.begin();
            posend = pSocketObj->m_MsgSendQueue.end();

            while(pos != posend)
            {
                pMsgBuf = (*pos);                          //拿到的每个消息都是 消息头+包头+包体【但要注意，我们是不发送消息头给客户端的】
                pMsgHeader = (LPSTRUC_MSG_HEADER)pMsgBuf;  //指向消息头
                pPkgHeader = (LPCOMM_PKG_HEADER)(pMsgBuf+pSocketObj->m_iLenMsgHeader);  //指向包头
                p_Conn = pMsgHeader->pConn;

                if(p_Conn->iCurrsequence != pMsgHeader->iCurrsequence) 
                {
                    pos2=pos;
                    pos++;
                    pSocketObj->m_MsgSendQueue.erase(pos2);
                    --pSocketObj->m_iSendMsgQueueCount; //发送消息队列容量少1 
                    p_memory->FreeMemory(pMsgBuf);  
                    continue;
                }

                if(p_Conn->iThrowsendCount > 0) 
                {
                    pos++;
                    continue;
                }

                --p_Conn->iSendCount;

                //走到这里，可以发送消息，一些必须的信息记录，要发送的东西也要从发送队列里干掉
                p_Conn->psendMemPointer = pMsgBuf;      //发送后释放用的，因为这段内存是new出来的
                pos2=pos;
                pos++;
                pSocketObj->m_MsgSendQueue.erase(pos2);
                --pSocketObj->m_iSendMsgQueueCount;      //发送消息队列容量少1   
                p_Conn->psendbuf = (char *)pPkgHeader;   //要发送的数据的缓冲区指针，因为发送数据不一定全部都能发送出去，我们要记录数据发送到了哪里，需要知道下次数据从哪里开始发送
                itmp = ntohs(pPkgHeader->pkgLen);        //包头+包体 长度 ，打包时用了htons【本机序转网络序】，所以这里为了得到该数值，用了个ntohs【网络序转本机序】；
                p_Conn->isendlen = itmp;                 //要发送多少数据，因为发送数据不一定全部都能发送出去，我们需要知道剩余有多少数据还没发送

                /*开始发送数据了*/
                sendsize = pSocketObj->sendproc(p_Conn,p_Conn->psendbuf,p_Conn->isendlen);
                if(sendsize > 0)
                {
                    if(sendsize == p_Conn->isendlen)
                    {
                        p_memory->FreeMemory(p_Conn->psendMemPointer);
                        p_Conn->psendMemPointer = NULL;
                        p_Conn->iThrowsendCount = 0;
                    }
                    else
                    {
                        /*数据没一次性发送完毕*/
                        p_Conn->psendbuf = p_Conn->psendbuf + sendsize;
                        p_Conn->isendlen = p_Conn->isendlen - sendsize;
                        /*标记发送缓冲区满了，需要用epoll来通知*/
                        ++p_Conn->iThrowsendCount;
                        if(pSocketObj->ngx_epoll_oper_event(
                                p_Conn->fd,         //socket句柄
                                EPOLL_CTL_MOD,      //事件类型
                                EPOLLOUT,           //标志，这里代表要增加的标志,EPOLLOUT：
                                0,                  //对于事件类型为增加的，EPOLL_CTL_MOD需要这个参数, 0：增加   1：去掉 2：完全覆盖
                                p_Conn              //连接池中的连接
                                ) == -1)
                        {
                            ngx_log_stderr(errno,"CSocekt::ServerSendQueueThread()\
                                ngx_epoll_oper_event()失败.");
                        }
                    }
                    continue;
                }
                else if(sendsize == 0)
                {
                    p_memory->FreeMemory(p_Conn->psendMemPointer);
                    p_Conn->psendMemPointer = NULL;
                    p_Conn->iThrowsendCount = 0;
                    continue;
                }
                else if(sendsize == -1)
                {
                    ++p_Conn->iThrowsendCount;
                    if(pSocketObj->ngx_epoll_oper_event(
                                p_Conn->fd,         //socket句柄
                                EPOLL_CTL_MOD,      //事件类型，
                                EPOLLOUT,           //标志，这里代表要增加的标志,EPOLLOUT：
                                0,                  //对于事件类型为增加的，EPOLL_CTL_MOD需要这个参数, 0：增加   1：去掉 2：完全覆盖
                                p_Conn              //连接池中的连接
                                ) == -1)
                    {
                        ngx_log_stderr(errno,"CSocekt::ServerSendQueueThread()\
                            中ngx_epoll_add_event()_2失败.");
                    }
                    continue;
                }
                else
                {
                    p_memory->FreeMemory(p_Conn->psendMemPointer); 
                    p_Conn->psendMemPointer = NULL;
                    p_Conn->iThrowsendCount = 0;
                }
            }
            err = pthread_mutex_unlock(&pSocketObj->m_sendMessageQueueMutex); 
            if(err != 0)
            {
                ngx_log_stderr(err,"CSocekt::ServerSendQueueThread()\
                    pthread_mutex_unlock()失败，返回的错误码为%d!",err);
            }
        }
    }
    return (void*)0;
}
/***************************************************************
 * @file       ngx_c_socket.cxx
 * @brief      打印统计信息
 * @author     txj
 * @version    v1
 * @return     
 * @date       2021/7/29
 **************************************************************/
void CSocekt::printTDInfo()
{
    time_t currtime = time(NULL);
    /*超过10秒我们打印一次*/
    if( (currtime - m_lastprintTime) > 10)
    {
        int tmprmqc = g_threadpool.getRecvMsgQueueCount();
        m_lastprintTime = currtime;
        int tmpoLUC = m_onlineUserCount;
        int tmpsmqc = m_iSendMsgQueueCount;
        ngx_log_stderr(0,"当前在线人数/总人数(%d/%d)。",tmpoLUC,m_worker_connections);        
        ngx_log_stderr(0,"连接池中空闲连接/总连接/要释放的连接(%d/%d/%d)。",m_freeconnectionList.size(),m_connectionList.size(),m_recyconnectionList.size());
        ngx_log_stderr(0,"当前时间队列大小(%d)。",m_timerQueuemap.size());        
        ngx_log_stderr(0,"当前收消息队列/发消息队列大小分别为(%d/%d)，丢弃的待发送数据包数量为%d。",tmprmqc,tmpsmqc,m_iDiscardSendPkgCount);        
        if( tmprmqc > 100000)
        {
            ngx_log_stderr(0,"接收队列条目数量过大(%d)，要考虑限速或者增加处理线程数量了！！！！！！",tmprmqc);
        }
    }
    return;
}